package com.bytecub.simulate.device.mqtt;

import com.bytecub.common.constants.BCConstants;
import com.bytecub.common.domain.gateway.direct.response.PropertyGetResponse;
import com.bytecub.common.domain.message.DeviceDownMessage;
import com.bytecub.simulate.device.instance.DeviceInstance;
import com.bytecub.utils.JSONProvider;
import org.eclipse.paho.client.mqttv3.*;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.List;

/**
 *  * ByteCub.cn.
 *  * Copyright (c) 2020-2021 All Rights Reserved.
 *  * 
 *  * @author bytecub@163.com songbin
 *  * @version Id: SimulateMqttCallback.java, v 0.1 2021-01-07  Exp $$
 *  
 */
public class SimulateMqttCallback implements MqttCallback {
    private static final Logger logger = LoggerFactory.getLogger(SimulateMqttCallback.class);
    private SimulateMqttClient client;
    private MqttConnectOptions options;
    private String urls;
    private List<String> topics;
    private DeviceInstance deviceInstance;

    public SimulateMqttCallback(DeviceInstance deviceInstance, SimulateMqttClient client, MqttConnectOptions options,
            String urls) {
        this.client = client;
        this.options = options;
        this.urls = urls;
        this.deviceInstance = deviceInstance;
    }

    @Override
    public void connectionLost(Throwable cause) {
        // 连接丢失后，一般在这里面进行重连
        logger.info("mqtt 连接丢失", cause);
        int count = 1;
        // int sleepTime = 0;
        boolean willConnect = true;
        while (willConnect) {
            try {
                Thread.sleep(100);
                logger.info("连接[{}]断开，尝试重连第{}次", this.client.getMqttClient().getServerURI(), count++);
                this.client.getMqttClient().connect(this.options);
                List<String> topics = client.getTopics();
                this.client.getMqttClient()
                        .subscribe((String[]) this.client.getTopics().toArray(new String[topics.size()]));
                logger.info("重连成功");
                willConnect = false;
            } catch (Exception e) {
                logger.warn("重连异常", e);
            }

        }

    }

    @Override
    public void deliveryComplete(IMqttDeliveryToken token) {
        // System.out.println("deliveryComplete---------" + token.isComplete());
    }

    @Override
    public void messageArrived(String topic, MqttMessage message) throws Exception {
        // subscribe后得到的消息会执行到这里面
        try {
            String msg = new String(message.getPayload(), BCConstants.GLOBAL.CHARSET_UFT8);
            // String msgGb = new String(message.getPayload(),
            // BCConstants.GLOBAL.CHARSET_GB2312);
            logger.info("UFT8消息 :{} \n  GB2312消息 :{}", msg, msg);
            DeviceDownMessage deviceDownMessage = JSONProvider.parseObject(msg, DeviceDownMessage.class);
            if (topic.endsWith("property/get")) {
                this.deviceInstance.PropertyReaderReply(deviceDownMessage.getIdentifier(),
                        deviceDownMessage.getMessageId());
            } else if (topic.endsWith("property/set") || topic.endsWith("service/invoke")) {
                this.deviceInstance.PropertyWriterReply(deviceDownMessage.getMessageId());
            } else if (topic.endsWith("/upgrade/set")) {
                // 设备升级
                this.deviceInstance.UpgradeReply(deviceDownMessage);
            }
            logger.info("发布消息客户端{}接收消息主题 : {}   消息内容: {}", client.getMqttClient().getServerURI(), topic, msg);
        } catch (Exception e) {
            logger.warn("mqtt 订阅消息异常", e);
        }

    }
}
